package dao

import (
	"bytes"
	"context"
	"database/sql"
	"runtime"
	"time"

	"github.com/nsqio/go-nsq"
	"gorm.io/gorm"

	"open/internal/model"
	"open/internal/types"
	"open/pkg/log"
)

// Transaction 通过ctx传递事务
func (d *Dao) Transaction(ctx context.Context, f func(ctx context.Context) error, opts ...*sql.TxOptions) (err error) {
	if err = d.db.Transaction(func(tx *gorm.DB) (err error) {
		events := make(map[string]map[int64][]byte)
		ctx = context.WithValue(ctx, eventKey, events)
		ctx = context.WithValue(ctx, transactionKey, tx)
		return f(ctx)
	}, opts...); err != nil {
		return
	}
	return d.pushEvent(ctx)

}

func (d *Dao) pushEvent(ctx context.Context) (err error) {
	if events, ok := ctx.Value(eventKey).(map[string]map[int64][]byte); ok {
		if len(events) != 0 {
			// 事务成功后，发送NSQ异步消息
			for topic, msgMap := range events {
				ch := make(chan *nsq.ProducerTransaction)
				msgIds := make([]int64, 0)
				msgBodies := make([][]byte, 0)
				for msgId, msgBody := range msgMap {
					msgIds = append(msgIds, msgId)
					msgBodies = append(msgBodies, msgBody)
				}
				logger := log.FromContext(ctx)

				go func() {
					ctx = context.Background()
					ctx = log.WithLogger(ctx, logger)
					defer func() {
						if rErr := recover(); rErr != nil {
							buf := make([]byte, 64<<10)
							n := runtime.Stack(buf, false)
							buf = buf[:n]
							logger.Errorf("%v: nsq pulish error \n%s\n", rErr, buf)
						}
					}()
					res := <-ch

					if res.Error != nil {
						if err = d.db.WithContext(ctx).Model(&model.EventMessage{}).
							Where("id in ? AND status = ?", msgIds, types.NSQInit).Updates(map[string]interface{}{
							"retry":        gorm.Expr("inv_nsq_message.retry + 1"),
							"execute_time": gorm.Expr("inv_nsq_message.execute_time + interval'1 minute' * inv_nsq_message.retry * inv_nsq_message.retry"),
							"update_time":  time.Now()}).Error; err != nil {
							log.FromContext(ctx).Errorf("nsq message update failed: %s", err)
						}
					} else {
						if err = d.db.WithContext(ctx).Model(&model.EventMessage{}).Where(
							"id in ? AND status = ?", msgIds, types.NSQInit).Updates(map[string]interface{}{
							"status": types.NSQSucceed}).Error; err != nil {
							log.FromContext(ctx).Errorf("nsq message update failed: %s", err)
						}
					}
					var msg string
					log.FromContext(ctx).Infow("nsq",
						"status_code", msg,
						"method", "publish",
						"path", topic,
						"body", string(bytes.Join(msgBodies, []byte(","))),
						"error", err,
					)
				}()
				if err = d.producer.MultiPublishAsync(topic, msgBodies, ch); err != nil {
					log.FromContext(ctx).Warnf("nsq send error: %s", err)
					// NSQ发送失败时，不抛出错误
					err = nil
				}
			}
		}
	}
	return
}
